package com.navinfo.platform.test;

import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import com.navinfo.opentsp.platform.location.protocol.gbv6.hangzhou.LCHangzhouDischarge;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class testKafkaClient {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.0.92:9092");//you should write specific ip address rather than localhost
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//StringSerializer/IntegerSerializer/or other self-defined Serializer.
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        properties.put("group.id", "MyGroup");
        properties.put("auto.offset.reset","earliest");

        Consumer<String,byte[]> consumer  = new KafkaConsumer<String,byte[]>(properties);
        consumer.subscribe(Arrays.asList("locationDataPB"));

        while(true){
            ConsumerRecords<String,byte[]> records = consumer.poll(100);
            for(ConsumerRecord re:records){
                try {
                    // D009
                    // LCIncidentAlarmData.IncidentAlarmData incidentAlarmData = LCIncidentAlarmData.IncidentAlarmData.parseFrom((byte[])(re.value()));
                    // D005
                    LCLocationData.LocationData locationData = LCLocationData.LocationData.parseFrom((byte[]) (re.value()));
                    Long tid = locationData.getTerminalId();
//                    LCHangzhouDischarge.HangzhouDischarge hangzhouDischarge = LCHangzhouDischarge.HangzhouDischarge.parseFrom((byte[]) (re.value()));
//                    Long tid = hangzhouDischarge.getTerminalId();
//                    System.out.println(hangzhouDischarge.toString());
//                    if(tid == 13300000015L){
//                      long tid1 =  tid;
//                    }
//
//                    Long gpstime =  hangzhouDischarge.getTime();

                }catch (Exception e){

                }
            }
        }
    }
}